package com.crazymakercircle.imServer.distributed;

import com.crazymakercircle.constants.ServerConstants;
import com.crazymakercircle.entity.ImNode;
import com.crazymakercircle.im.common.bean.msg.ProtoMsg;
import com.crazymakercircle.imServer.protoBuilder.NotificationMsgBuilder;
import com.crazymakercircle.util.JsonUtil;
import com.crazymakercircle.util.ObjectUtil;
import com.crazymakercircle.util.ThreadUtil;
import com.crazymakercircle.zk.CuratorZKclient;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;

import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/*

 WorkerRouter 是一个工作路由组件，主要用于在分布式系统中管理与其他节点（如工作节点）的通信和连接。它的主要功能包括：
1.  单例管理：通过 getInst() 方法提供全局唯一的 WorkerRouter 实例。
2.  ZooKeeper 集成：使用 CuratorFramework 客户端与 ZooKeeper 交互，监听路径 /MANAGE_PATH 下子节点的变化（添加、删除、更新），
* 以动态调整工作节点的状态。
3.  工作节点映射表 (workerMap)：
  1)存储每个工作节点 ID 到对应的 PeerSender 对象之间的映射。
  2)PeerSender 负责与特定的工作节点进行网络通信。
4.  事件处理：
  1) 当有新的工作节点加入时 (CHILD_ADDED)，会触发 processNodeAdded 方法，建立或更新到该节点的连接。
  2) 当有工作节点移除时 (CHILD_REMOVED)，会触发 processNodeRemoved 方法，关闭并移除相应的连接。
5.  消息转发：提供 route(long nodeId) 方法来根据节点 ID 获取对应的 PeerSender，从而实现消息的定向发送。
6.  通知广播：利用 sendNotification(String json) 方法向所有非本地的工作节点广播通知。
 **/
@Data
@Slf4j
public class WorkerRouter {
    //Zk客户端：用于与ZK进行交互
    private CuratorFramework client = null;
   //在ZK上注册的路径
    private String pathRegistered = null;
    //当前节点的信息
    private ImNode node = null;

    //WorkerRouter的单例实例
    private static WorkerRouter singleInstance = null;/*是一个工作路由组件，
    主要用于在分布式系统中管理与其他节点（如工作节点）的通信和连接*/


    /*所有节点管理的路径*/
    private static final String path = ServerConstants.MANAGE_PATH;
    /*工作节点映射，键 为工作节点的ID， 值为PeerSender对象，用于与工作节点进行通信*/
    private ConcurrentHashMap<Long, PeerSender> workerMap =
            new ConcurrentHashMap<>();

/*
在节点添加后执行的操作
 使用BiConsumer定义了一个当节点添加完成后需要执行的操作
*@param node ImNode类型，表示IM网络中的一个节点
 * @param relaySender PeerSender类型，用于发送消息到特定节点的发送器
 */

   private BiConsumer<ImNode, PeerSender> runAfterAdd = (node, relaySender) -> {
        doAfterAdd(node, relaySender);
    };

   /*在节点移除后执行的操作
    * 使用Consumer定义了一个当节点移除完成后需要执行的操作
    * @param node ImNode类型，表示IM网络中的一个节点
    * */
    private  Consumer<ImNode> runAfterRemove = (node) -> {
        doAfterRemove(node);
    };


    /* 获取WorkerRouter的单例实例
     * 该方法确保了在多线程环境下的安全访问和延迟初始化
     * @return WorkerRouter实例，保证在整个应用中只有一个实例
     **/
    public synchronized static WorkerRouter getInst() {
        if (null == singleInstance) {
            singleInstance = new WorkerRouter();
        }
        return singleInstance;
    }

    private WorkerRouter() {

    }

    private boolean inited=false;

    /**
     * 初始化节点管理
     */
    public void init() {

        if(inited)  //避免重复初始化
        {
            return;
        }
        inited=true;

        try {
            /*检查并实例化Zookeeper客户端*/
            if (null == client) {
                this.client = CuratorZKclient.instance.getClient();

            }

            //订阅节点的增加和删除事件
            PathChildrenCache childrenCache = new PathChildrenCache(client, path, true);
            PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {

                @Override
                public void childEvent(CuratorFramework client,
                                       PathChildrenCacheEvent event) throws Exception {
                    log.info("开始监听其他的ImWorker子节点:-----");/*日志记录，开始监听子节点变化*/
                    ChildData data = event.getData();
                    /*根据不同的事件类型处理节点变化*/
                    switch (event.getType()) {
                        case CHILD_ADDED: /*当子节点添加时*/
                            log.info("CHILD_ADDED : " + data.getPath() + "  数据:" + data.getData());
                            processNodeAdded(data);
                            break;
                        case CHILD_REMOVED: /*当节点移除时*/
                            log.info("CHILD_REMOVED : " + data.getPath() + "  数据:" + data.getData());
                            processNodeRemoved(data);
                            break;
                        case CHILD_UPDATED: /*当节点更新时*/
                            log.info("CHILD_UPDATED : " + data.getPath() + "  数据:" + new String(data.getData()));
                            break;
                        default:  /*其他情况*/
                            log.debug("[PathChildrenCache]节点数据为空, path={}", data == null ? "null" : data.getPath());
                            break;
                    }

                }

            };
            /*注册监听器 并 启动缓存*/
            childrenCache.getListenable().addListener(
                    childrenCacheListener, ThreadUtil.getIoIntenseTargetThreadPool());
            System.out.println("Register zk watcher successfully!");
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        } catch (Exception e) {
            /*异常堆栈打印信息*/
            e.printStackTrace();
        }
    }

    /*处理节点被移除的情况
     * 当监视的节点被移除时，此方法将被调用，它负责执行一些清理工作，
     * 并通知感兴趣的监听器该节点已不再可用*/
    private void processNodeRemoved(ChildData data) {

        byte[] payload = data.getData();/*将节点数据转换 为 ImNode对象*/
        ImNode node = ObjectUtil.JsonBytes2Object(payload, ImNode.class);

        /*获取节点ID，并设置到ImNode对象 中*/
        long id = ImWorker.getInst().getIdByPath(data.getPath());
        node.setId(id);
        /*记录节点删除的信息*/
        log.info("[TreeCache]节点删除, path={}, data={}",
                data.getPath(), JsonUtil.pojoToJson(node));

        /*如果有移除节点后的处理逻辑，则执行之  */
        if (runAfterRemove != null) {
            runAfterRemove.accept(node);
        }


    }
    /*
     * 在节点移除后执行的操作
     * 此方法旨在停止与移除节点相关的PeerSender的连接，并从workerMap中移除该节点
     *
     * @param node 表示被移除的节点的ImNode对象
     */
    private void doAfterRemove(ImNode node) {
        PeerSender peerSender = workerMap.get(node.getId());/*获取与节点ID关联的PeerSender实例*/

        if (null != peerSender) {  /*如果PeerSender实例 存在
        ，则停止其连接，并从workerMap中移除对应节点*/
            peerSender.stopConnecting();
            workerMap.remove(node.getId());
        }


    }

    /*
     *
     *
     * 处理节点添加事件
     * 当监控到ZooKeeper中有一个新的子节点被添加时，该方法会被调用
     *
     * @param data 包含被添加节点信息的ChildData对象
     */
    private void processNodeAdded(ChildData data) {
        /*将节点数据转换为字节数组*/
              byte[] payload = data.getData();

        ImNode node = ObjectUtil.JsonBytes2Object(payload, ImNode.class);/*将字节数组反序列化为ImNode对象*/

        long id = ImWorker.getInst().getIdByPath(data.getPath());/*通过节点路径获取节点ID*/
        node.setId(id); /*设置节点ID*/

        log.info("[TreeCache]节点更新端口, path={}, data={}",
                data.getPath(), JsonUtil.pojoToJson(node));/*日志记录——节点更新信息*/

        if (node.equals(getLocalNode())) {
            log.info("[TreeCache]本地节点, path={}, data={}",
                    data.getPath(), JsonUtil.pojoToJson(node));/*如果新添加的节点是本地节点，则记录日志并返回*/
            return;
        }
        PeerSender relaySender = workerMap.get(node.getId());/*获取与节点ID关联的PeerSender对象*/
        //重复收到注册的事件
        if (null != relaySender && relaySender.getRmNode().equals(node)) {/*如果节点重复添加，则记录日志并返回*/

            log.info("[TreeCache]节点重复增加, path={}, data={}",
                    data.getPath(), JsonUtil.pojoToJson(node));
            return;
        }

        if (runAfterAdd != null) {   /*如果存在添加节点后的操作，则执行该操作*/
            runAfterAdd.accept(node, relaySender);
        }
    }

    /*
     * 在添加节点后执行的操作
     *
     * @param n 新增的节点信息
     * @param relaySender 当前节点对应的PeerSender对象，用于数据转发
     */
    private void doAfterAdd(ImNode n, PeerSender relaySender) {
        if (null != relaySender) {
            //关闭老的连接
            relaySender.stopConnecting();
        }
        //创建一个消息转发器
        relaySender = new PeerSender(n);
        //建立转发的连接
        relaySender.doConnect();

        workerMap.put(n.getId(), relaySender);/*将worker的ID作为键，对应的relaySender作为值存入workerMap中*/
    }

/* 根据节点ID获取对应的PeerSender对象
 *
 * @param nodeId 节点ID
 * @return 返回对应的PeerSender对象，如果不存在则返回null*/
    public PeerSender route(long nodeId) {
        PeerSender peerSender = workerMap.get(nodeId);
        if (null != peerSender) {
            return peerSender;
        }
        return null;
    }

    /*
     * 向所有其他节点发送通知消息
     *
     * @param json 通知消息的内容，格式为JSON字符串
     */
    public void sendNotification(String json) {
        workerMap.keySet().stream().forEach(
                key ->
                {
                    if (!key.equals(getLocalNode().getId())) {
                        PeerSender peerSender = workerMap.get(key);
                        ProtoMsg.Message pkg = NotificationMsgBuilder.buildNotification(json);
                        peerSender.writeAndFlush(pkg);
                    }
                }
        );

    }

    /*
     * 获取本地节点信息
     *
     * @return 返回本地节点信息对象
     */
    public ImNode getLocalNode() {
        return ImWorker.getInst().getLocalNodeInfo();
    }

    /*
     * 移除指定的远程节点信息
     *
     * @param remoteNode 要移除的远程节点信息
     */
    public void remove(ImNode remoteNode) {
        workerMap.remove(remoteNode.getId());
        log.info("[TreeCache]移除远程节点信息,  node={}", JsonUtil.pojoToJson(remoteNode));
    }
}
